
#include "thread/executor.h"
#include <algorithm>
#include <unistd.h>
#include "base/logging.h"
#include "base/callback.h"
#include "base/cmd_flags.h"
#include "base/logging.h"
#include "base/mutex.h"
#include "util/time.h"
#include "thread/timer.h"

using std::map;
using std::min;
using std::vector;

DEFINE_int32(default_threads, 4, "");

//Executor* Executor::default_exectuor_ = NULL;

CallbackThread::CallbackThread(Closure* callback)
    : callback_(callback), done_callback_(NULL) {
}

void CallbackThread::set_done_callback(Closure* done_callback) {
  done_callback_ = done_callback;
}

void CallbackThread::Run() {
  callback_->Run();
  if (done_callback_ != NULL)  {
    done_callback_->Run();
  }
}

/*
Executor* Executor::GetDefault() {
  if (default_exectuor_ == NULL) {
    default_exectuor_ = new DefaultExecutor(FLAGS_default_threads);
  }
  return default_exectuor_;
}*/

DefaultExecutor::DefaultExecutor(int num_threads)
    : finalize_mutex_(PTHREAD_MUTEX_INITIALIZER),
      finalize_cond_(PTHREAD_COND_INITIALIZER),
      num_threads_(num_threads),
      current_id_(0),
      waiting_to_be_finalized_(false) {
}

DefaultExecutor::~DefaultExecutor() {
  CHECK_EQ(pthread_mutex_lock(&finalize_mutex_), 0);
  if (pending_callbacks_.size() != 0 ||
      running_callbacks_.size() != 0 || 
      pending_timed_callbacks_.size() != 0) {
    waiting_to_be_finalized_ = true;
    pthread_cond_wait(&finalize_cond_, &finalize_mutex_);
  }
  CHECK_EQ(pthread_mutex_unlock(&finalize_mutex_), 0);
}

int64 DefaultExecutor::Add(Closure* callback) {
  CallbackState* new_state = new CallbackState();
  new_state->callback = callback;
  new_state->thread = new CallbackThread(callback);
  new_state->thread->set_done_callback(NewCallback(this, &DefaultExecutor::MarkDone, new_state));
  {
    WriteLock lock(&mutex_);
    pending_callbacks_[current_id_] = new_state;
    new_state->id = current_id_;
    ++current_id_;
  }

  StartRunnableClosure();

  return new_state->id;
}

int64 DefaultExecutor::AddAfter(int64 usec, Closure* callback) {
  CallbackState* new_state = new CallbackState();
  new_state->callback = callback;
  new_state->thread = new CallbackThread(callback);
  new_state->thread->set_done_callback(NewCallback(this, &DefaultExecutor::MarkDone, new_state));
  {
    WriteLock lock(&mutex_);
    pending_timed_callbacks_[GetCurrentTime() + usec][current_id_] = new_state;
    new_state->id = current_id_;
    ++current_id_;
  }
  Timer::AddAfter(usec, NewCallback(this, &DefaultExecutor::MovePendingTimedClosure));

  return new_state->id;
} 

bool DefaultExecutor::Cancel(int64 id) {
  {
    WriteLock lock(&mutex_);
    if (pending_callbacks_.find(id) != pending_callbacks_.end()) {
      pending_callbacks_.erase(id);
      return true;
    }
  }
  return false;
}
  
void DefaultExecutor::MarkDone(CallbackState* state) {
  state->thread->Join();
  {
    WriteLock lock(&mutex_);
    running_callbacks_.erase(state->id);
  }
  delete state->thread;
  delete state;
  CHECK_EQ(pthread_mutex_lock(&finalize_mutex_), 0);
  if (waiting_to_be_finalized_ &&
      pending_callbacks_.size() == 0 &&
      running_callbacks_.size() == 0) {
    pthread_cond_signal(&finalize_cond_);
  }
  CHECK_EQ(pthread_mutex_unlock(&finalize_mutex_), 0);
  StartRunnableClosure();
}

void DefaultExecutor::MovePendingTimedClosure() {
  {
    WriteLock lock(&mutex_);
    while (true) {
      if (pending_timed_callbacks_.size() > 0 &&
          pending_timed_callbacks_.begin()->first <= GetCurrentTime()) {
        int64 t = pending_timed_callbacks_.begin()->first;
        map<int64, CallbackState*>* current_callbacks = &pending_timed_callbacks_.begin()->second;
        CHECK_GT(current_callbacks->size(), 0);
        int64 id = current_callbacks->begin()->first;
        CallbackState* state = current_callbacks->begin()->second;
        current_callbacks->erase(id);
        if (current_callbacks->size() == 0) {
          pending_timed_callbacks_.erase(t);
        }
        pending_callbacks_[id] = state;
        continue;
      } else {
        break;
      }
    }
  }
  StartRunnableClosure();
}
 
void DefaultExecutor::StartRunnableClosure() {
  WriteLock lock(&mutex_);
  if (num_threads_ <= running_callbacks_.size()) {
    return;
  }
  while (num_threads_ > running_callbacks_.size()) {
    if (pending_callbacks_.size() == 0) {
      break;
    }
    if (pending_callbacks_.size() > 0) {
      int64 id = pending_callbacks_.begin()->first;
      CallbackState* state = pending_callbacks_.begin()->second;
      pending_callbacks_.erase(id);
      running_callbacks_[id] = state;
      state->thread->Start();
      continue;
    }
  }
}
